package net.wicp.tams.common.binlog.alone.parser;

import java.io.IOException;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;

import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.ArrayUtils;

import lombok.extern.slf4j.Slf4j;
import net.wicp.tams.common.apiext.CollectionUtil;
import net.wicp.tams.common.apiext.StringUtil;
import net.wicp.tams.common.apiext.TimeAssist;
import net.wicp.tams.common.apiext.jdbc.JdbcAssit;
import net.wicp.tams.common.apiext.jdbc.JdbcConnection;
import net.wicp.tams.common.binlog.alone.BusiAssit;
import net.wicp.tams.common.binlog.alone.ListenerConf.CheckPoint;
import net.wicp.tams.common.binlog.alone.ListenerConf.ColHis;
import net.wicp.tams.common.binlog.alone.ListenerConf.ConnConf;
import net.wicp.tams.common.binlog.alone.ListenerConf.ConnConf.Builder;
import net.wicp.tams.common.binlog.alone.ListenerConf.Position;
import net.wicp.tams.common.binlog.parser.DirectLogFetcher;
import net.wicp.tams.common.binlog.parser.LogContext;
import net.wicp.tams.common.binlog.parser.LogDecoder;
import net.wicp.tams.common.binlog.parser.LogEvent;
import net.wicp.tams.common.binlog.parser.event.DeleteRowsLogEvent;
import net.wicp.tams.common.binlog.parser.event.GtidLogEvent;
import net.wicp.tams.common.binlog.parser.event.QueryLogEvent;
import net.wicp.tams.common.binlog.parser.event.RotateLogEvent;
import net.wicp.tams.common.binlog.parser.event.UpdateRowsLogEvent;
import net.wicp.tams.common.binlog.parser.event.WriteRowsLogEvent;
import net.wicp.tams.common.binlog.parser.event.XidLogEvent;
import net.wicp.tams.common.constant.DateFormatCase;
import net.wicp.tams.common.constant.OptType;
import net.wicp.tams.common.thread.threadlocal.PerthreadManager;

@Slf4j
public class ParseLogOnline extends BaseLogFetcher {

	private Connection conn;
	private String uuid;
	private String slaveGtids;// 从服务器第一个字符为","

	private boolean gtidCan;
	// 在做主备切换时会有此值
	private Position posPre;

	private DirectLogFetcher fecther;
	// 当前位点
	private Position.Builder curpos;
	private Position.Builder savepos;// 可以被保存的最后位点
	ScheduledExecutorService timerService;
	private final String configName;

	public static List<Thread> startThreadList = new ArrayList<Thread>();

	public ParseLogOnline(ConnConf.Builder connConfBuilder) {
		super(connConfBuilder);
		this.configName = connConfBuilder.getConfName();
		startThreadList.add(Thread.currentThread());
		addTimer();
	}

	@Override
	protected void init(Builder connConfBuilder) {
		switch (connConfBuilder.getHaType()) {
		case cur:
			connConfBuilder.setPos(getMastStatus(connConfBuilder));
			break;
		case pos:
			if (!connConfBuilder.hasPos()) {
				throw new RuntimeException("hatype为pos类型，却没有提供gtid等位置信息。");
			}
			break;
		case last:
			Position maxpoint = saveCheckPoint.findPoint(Long.MAX_VALUE);
			if (maxpoint == null) {// 第一次启动退化为从当前位点启动
				connConfBuilder.setPos(getMastStatus(connConfBuilder));
			} else {
				connConfBuilder.setPos(maxpoint);
			}
			break;
		default:
			throw new RuntimeException("不支持的hatype。");
		}

		/*
		 * if (!connConfBuilder.hasPos()) {// 当前目录
		 * connConfBuilder.setPos(getMastStatus(connConfBuilder)); }
		 */
		this.curpos = connConfBuilder.getPosBuilder().clone();
		try {
			if (this.conn != null && !this.conn.isClosed()) {
				this.conn.close();
			}
		} catch (Exception e) {
			log.error("关闭链接失败", e);
		}
		try {
			Class.forName("com.mysql.jdbc.Driver");
			this.conn = DriverManager.getConnection(
					String.format("jdbc:mysql://%s:%s", connConfBuilder.getIp(), connConfBuilder.getPort()),
					connConfBuilder.getUser(), connConfBuilder.getPassword());

			Statement statement = this.conn.createStatement();
			statement.execute("SET @master_binlog_checksum='@@global.binlog_checksum'");
			statement.execute("SET @mariadb_slave_capability='" + LogEvent.MARIA_SLAVE_CAPABILITY_MINE + "'");
			statement.close();

			uuid = getVar(this.conn, "server_uuid", false);
			gtids = connConfBuilder.getPos().getGtids();
			super.fileName = connConfBuilder.getPos().getFileName();
			// 设置一些参数
			// this.checksum = Checksum.get(getVar(conn, "binlog_checksum", false));
			initSlaveGtids(connConfBuilder.getPos().getGtids());
			gtidCan = isAvailable(this.conn, connConfBuilder.getPos().getGtids(), uuid);// 可以使用gtid
			log.info("gtid：[{}]是否可用：[{}]", connConfBuilder.getPos().getGtids(), gtidCan);
		} catch (Exception e) {
			log.error("读binlog日志出现问题", e);
			throw new RuntimeException(e);
		}

	}

	ScheduledFuture<?> scheduleAtFixedRate = null;

	private void addTimer() {
		timerService = Executors.newSingleThreadScheduledExecutor();
		// 第二个参数为首次执行的延时时间，第三个参数为定时执行的间隔时间
		scheduleAtFixedRate = timerService.scheduleAtFixedRate(new Runnable() {
			@Override
			public void run() {

				try {
					savepos = savepos == null ? curpos.clone() : savepos;// 防止因为第一次一直没有触发而导致位点一直不走的现象
					if (savepos != null) {
						Position pos = savepos.build();
						saveCheckPoint.savePoint(pos);
					} else {
						log.info("configName:{} pos is null", configName);
					}
				} catch (Throwable e) {
					log.error("save point error:", e);
				}
			}
		}, 10, 10, TimeUnit.SECONDS);
	}

	private int max = 8;// 256秒即4分种*2=8分钟，8分钟就不是网络抖动问题了

	private boolean isClose = false;

	public boolean isClose() {
		return this.isClose;
	}

	/***
	 * 需要考虑网络斗动问题
	 */
	@Override
	public void read() {
		int sendNum = 0;
		int timeout = 1;
		while (true) {
			if (isClose) {
				break;
			}
			if (sendNum >= max) {
				log.error("网络出现问题，需要联系相关人员");
				break;
			}
			try {
				readDo();
			} catch (Throwable e) {
				log.error("readdo error", e);
			} finally {
				if (!isClose) {
					PerthreadManager.getInstance().createValue("need-init", Boolean.class).set(true);
					sendNum++;
					try {
						Thread.sleep(timeout * 1000);
					} catch (InterruptedException e) {
					}
					log.error("第[{}]次发送失败，等待时间：[{}].请联系相关人员。", sendNum, timeout);
					timeout = timeout * 2;
				}
			}
		}
	}

	private void readDo() {
		if (StringUtil.isNull(super.connConf.getPos().getGtids()) && super.connConf.getPos().getPos() < 4) {// 想通过位置方式
			throw new IllegalAccessError("开始位置最小为4");
		}

		Boolean needinit = PerthreadManager.getInstance().createValue("need-init", Boolean.class).get(false);
		if (needinit) {
			init(connConf.toBuilder());
		}
		fecther = new DirectLogFetcher();
		try {
			fetchLog();
			//
			LogDecoder decoder = new LogDecoder(LogEvent.UNKNOWN_EVENT, LogEvent.ENUM_END_EVENT);
			LogContext context = new LogContext();
			boolean isSel = false;
			while (fecther.fetch()) {
				metric.meter_parser_pack_all.mark();// 总包数
				LogEvent event = null;
				event = decoder.decode(fecther, context);
				if (event == null) {
					continue;
					// throw new RuntimeException("parse failed");
				}

				int eventType = event.getHeader().getType();
				switch (eventType) {
				case LogEvent.FORMAT_DESCRIPTION_EVENT: // MySQL
														// Server的版本，binlog的版本，该binlog文件的创建时间
					// 第二个解析事件，暂不做任何事
					break;
				case LogEvent.ROTATE_EVENT:// 第一个解析事件，需要它提供potion，否则报空指针
					super.fileName = ((RotateLogEvent) event).getFilename();
					break;
				case LogEvent.WRITE_ROWS_EVENT_V1:
				case LogEvent.WRITE_ROWS_EVENT:
					if (super.gtids != null && parseRowsEvent((WriteRowsLogEvent) event, OptType.insert)) {
						isSel = true;
						this.savepos = this.curpos.clone();
					} else {// 非监听的变化，需要使用this.curpos当前位点处理，设置savepos=null;在定时器执行时会使用当前位点。
						if (!isChkDb) {// 需要忽略checkpoint的修改，否则会死循环，不断插入point，不断保存最新位点。如果设置它时replace旧位点不会产生binlog
							this.savepos = null;
						}
					}
					break;
				case LogEvent.UPDATE_ROWS_EVENT_V1:
				case LogEvent.UPDATE_ROWS_EVENT:
					if (super.gtids != null && parseRowsEvent((UpdateRowsLogEvent) event, OptType.update)) {
						isSel = true;
						this.savepos = this.curpos.clone();
					} else {
						if (!isChkDb) {
							this.savepos = null;
						}
					}
					break;
				case LogEvent.DELETE_ROWS_EVENT_V1:
				case LogEvent.DELETE_ROWS_EVENT:
					if (super.gtids != null && parseRowsEvent((DeleteRowsLogEvent) event, OptType.delete)) {
						isSel = true;
						this.savepos = this.curpos.clone();
					} else {
						if (!isChkDb) {
							this.savepos = null;
						}
					}
					break;
				case LogEvent.QUERY_EVENT:
					parseQueryEvent((QueryLogEvent) event);
					break;
				case LogEvent.XID_EVENT:
					if (super.gtids != null && isSel) {
						parseXidEvent((XidLogEvent) event);
						isSel = false;
					}
					break;
				case LogEvent.GTID_LOG_EVENT:
					parseGtidLogEvent((GtidLogEvent) event);
					break;
				default:
					break;
				}
			}
		} catch (SQLException e) {// 得到联接错误
			log.error("得到联接错误", e);
		} catch (IOException e) {// 拉取日志文件错误
			if (e.getMessage().contains("errno = 1236, sqlstate = HY000")) {
				// TODO 偿试处理要处理的位点被删除的错误
			}
			log.error("拉取日志文件错误", e);
		} catch (Throwable e) {// 其它错误
			log.error("未知错误", e);
		} finally {
			// 不能关闭，因为fecther关闭会关掉conn,共用同一个。
			/*
			 * try { fecther.close(); } catch (IOException e) { log.error("关闭fecther失败", e);
			 * }
			 */
		}
	}

	// 添加重连机制
	private Connection getConn() {
		while (true) {
			try {
				if (this.conn == null || this.conn.isClosed()) {// 如果连接意外销毁
					init(connConf.toBuilder());
				}
				TimeAssist.reDoWaitInit("tams-binlog");
				return this.conn;
			} catch (Exception e) {
				log.error("得到连接失败", e);
				boolean reDoWait = TimeAssist.reDoWait("tams-binlog", 5);
				if (reDoWait) {
					try {
						Thread.sleep(32000L);// 到达了最大值,一直等待
					} catch (InterruptedException e1) {
					}
				}
			}
		}
	}

	private void fetchLog() throws IOException, CloneNotSupportedException {
		boolean useGtid = true;
		boolean needSecond = false;

		try {
			if (gtidCan) {
				fecther.openGtid(getConn(), gtids, super.connConf.getClientId());
			} else {
				useGtid = false;
				fetchLogFroPos();
			}
		} catch (Exception e) {
			log.error("第一次启动失败，使用gtid:" + useGtid, e);
			needSecond = true;
		}
		if (needSecond) {// 偿试第二次换种方式启动
			if (useGtid) {
				fetchLogFroPos();
			} else {
				fecther.openGtid(getConn(), gtids, super.connConf.getClientId());
			}
		}
	}

	private void fetchLogFroPos() throws IOException {
		long filePosition = super.connConf.getPos().getPos();
		if (this.posPre != null) {
			int upPos = 10000;
			filePosition = super.connConf.getPos().getPos() > upPos + 4 ? (super.connConf.getPos().getPos() - upPos)
					: super.connConf.getPos().getPos();
			log.warn("已做主备切换，但不支持Gtid，采用回朔10000位点的方式");
		}
		fecther.open(getConn(), super.connConf.getPos().getFileName(), filePosition, super.connConf.getClientId());
	}

	@Override
	protected void parseGtidLogEventSub(GtidLogEvent event) {
		if (StringUtil.isNotNull(uuid) && uuid.equals(event.getSource())) {// 当做主备时会变化source
			if (StringUtil.isNotNull(slaveGtids)) {
				super.gtids = String.format("%s,%s", super.gtids, slaveGtids);
			}
			// 设置解析的位点
			this.curpos.setGtids(super.gtids);
			this.curpos.setPos(event.getLogPos());
			long time = event.getHeader().getWhen() * 1000;
			this.curpos.setTime(time);
			this.curpos.setTimeStr(DateFormatCase.YYYY_MM_DD_hhmmss.getInstanc().format(time));
			this.curpos.setServerIp(super.connConf.getIp());
			this.curpos.setClintId(super.connConf.getClientId());
		} else {
			log.info("------------------做主备切换,原主机源[{}],切换源[{}}]--------------------------------", uuid,
					event.getSource());
			super.gtids = null;
		}
	}

	public Position getCurpos() {
		// 当运行初期没有数据进来时savepos可能为空，那就用初始的位点
		return this.savepos == null ? this.curpos.build() : this.savepos.build();
	}

	public CheckPoint getCheckPointCur() {
		Position curpos = getCurpos();
		return getCheckPoint(curpos);
	}

	public CheckPoint getCheckPoint(long time) {
		Position findTime = saveCheckPoint.findPoint(time);
		return getCheckPoint(findTime);
	}

	public void setColHis(List<ColHis> colHisList) {
		if (CollectionUtils.isEmpty(colHisList)) {
			return;
		}
		for (ColHis colHis : colHisList) {
			String key = BusiAssit.getColHiskey(colHis.getDb(), colHis.getTb());
			if (super.colsMap.containsKey(key)) {
				// 去重
				List<ColHis> list = super.colsMap.get(key);
				boolean has = false;
				for (ColHis ele : list) {
					if (ele.getTime() == colHis.getTime()) {
						has = true;
						break;
					}
				}
				if (!has) {
					list.add(colHis);
				}
			} else {
				List<ColHis> templist = new ArrayList<>();
				templist.add(colHis);
				super.colsMap.put(key, templist);
			}
		}
		// 保存h2db
		for (ColHis colHis2 : colHisList) {
			super.saveCheckPoint.saveColName(colHis2);
		}
		// 排序
		for (String key : super.colsMap.keySet()) {
			if (super.colsMap.get(key).size() >= 2) {
				Collections.sort(super.colsMap.get(key), new Comparator<ColHis>() {
					@Override
					public int compare(ColHis o1, ColHis o2) {
						long def = o2.getTime() - o1.getTime();
						return def > 0 ? 1 : (def < 0 ? -1 : 0);
					}
				});
			}
		}
	}

	private CheckPoint getCheckPoint(Position findPoint) {
		List<ColHis> colsList = saveCheckPoint.findColsAll();
		CheckPoint.Builder checkPointBuilder = CheckPoint.newBuilder();
		checkPointBuilder.setPos(findPoint);
		checkPointBuilder.addAllCols(colsList);
		return checkPointBuilder.build();
	}

	@Override
	public void close() {
		this.isClose = true;
		// 关闭的时候保存一下最后位点，相关于shutdown
		if (savepos != null) {
			saveCheckPoint.savePoint(savepos.build());
		}
		try {
			if (scheduleAtFixedRate != null) {
				scheduleAtFixedRate.cancel(true);
			}
			log.info("============1close timerService Thread sucess");
			if (timerService != null) {
				timerService.shutdown();
				log.info("============2close timerService pool sucess");
			}
		} catch (Throwable e) {
			log.error("关闭定时器失败", e);
		}

		if (super.getBinlogListener() != null) {
			try {
				super.getBinlogListener().close();
				log.info("============3close BinlogListener2 sucess");
			} catch (Throwable e) {
				log.error("关闭监听者失败", e);
			}
		}
		if (fecther != null) {
			try {
				fecther.close();
				log.info("============4close fecther sucess");
			} catch (Throwable e) {
				log.error("关闭fecther失败", e);
			}
		}
		try {
			saveCheckPoint.releaseLock();
		} catch (Exception e) {
			log.error("释放分布式锁失败", e);
		}
		// LoggerUtil.exit(JvmStatus.s15);线程模式，不能关闭
	}

	/////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////

	public String getVar(Connection conn, String var, boolean isGlobal) throws SQLException {
		String sql = String.format("select @@%s%s", isGlobal ? "GLOBAL." : "", var);
		ResultSet resultSet = JdbcAssit.querySql(conn, sql);
		resultSet.next();
		String retstr = resultSet.getString(1);
		resultSet.close();
		return retstr;
	}

	public boolean isAvailable(Connection conn, String gtidStr, String uuid) throws SQLException {
		if (StringUtil.isNull(gtidStr)) {// 都没传gtid，说明不走gtid
			return false;
		}
		long maxGtidDel = maxGtid(getVar(conn, "GTID_PURGED", true), uuid);
		long maxGtidCur = maxGtid(gtidStr.replace("\n", ""), uuid);
		if (maxGtidDel == 0 || (maxGtidDel > 0 && maxGtidCur > 0 && maxGtidCur > maxGtidDel)) {
			return true;
		} else {
			return false;
		}
	}

	private long maxGtid(String gtidStr, String uuid) {
		if (StringUtil.isNull(gtidStr)) {// 本机在@@GLOBAL.GTID_PURGED为“”的情况
			return 0;
		}
		String[] gtidAry = gtidStr.split(",");
		long delmax = 0;
		for (String eleGtid : gtidAry) {
			if (eleGtid.startsWith(uuid)) {
				int index1 = eleGtid.lastIndexOf(":");
				String[] nums = eleGtid.substring(index1 + 1).split("-");
				String delNumStr = nums.length > 1 ? nums[1] : nums[0];
				delmax = Long.parseLong(delNumStr);
				break;
			}
		}
		return delmax;
	}

	private void initSlaveGtids(String gtids) {
		if (StringUtil.isNotNull(gtids)) {
			this.gtids = gtids.replace("\n", "");
			String[] gtidsAry = this.gtids.split(",");

			for (String gtid : gtidsAry) {
				if (gtid.startsWith(uuid)) {
					gtidsAry = (String[]) ArrayUtils.removeElement(gtidsAry, gtid);
					break;
				}
			}
			this.slaveGtids = CollectionUtil.arrayJoin(gtidsAry, ",");
		}
	}

	private Position getMastStatus(Builder connConfBuilder) {
		String url;
		url = String.format("jdbc:mysql://%s:%s?autoReconnect=true&useUnicode=true&characterEncoding=utf-8",
				connConfBuilder.getIp(), connConfBuilder.getPort());
		java.sql.Connection conn = JdbcConnection.getConnection("com.mysql.jdbc.Driver", url, connConfBuilder.getUser(),
				connConfBuilder.getPassword());
		ResultSet rs = JdbcAssit.querySql(conn, "show master status");
		try {
			if (rs.next()) {
				String filename = rs.getString(1);
				long pos = rs.getLong(2);
				Position.Builder ret = Position.newBuilder();
				ret.setFileName(filename);
				ret.setPos(pos);
				if (rs.getMetaData().getColumnCount() >= 5) {
					String gtidStr = rs.getString(5);
					ret.setGtids(gtidStr.replace("/n", ""));
				}
				rs = JdbcAssit.querySql(conn, "show variables like 'server_id'");
				if (rs.next()) {
					long masterServerId = rs.getLong(2);
					ret.setMasterServerId(masterServerId);
				}
				rs = JdbcAssit.querySql(conn, "SELECT unix_timestamp(now())");// 得到时间戳，单位为秒
				if (rs.next()) {
					ret.setTime(rs.getLong(1) * 1000);
				}
				return ret.build();
			}
			throw new RuntimeException("没有得到mastStatus,服务器不支持binlog");
		} catch (SQLException e) {
			throw new RuntimeException(e);
		} finally {
			try {
				rs.close();
				conn.close();
			} catch (SQLException e) {
				e.printStackTrace();
			}
		}
	}

}
